package com.glodon.paas.framework.registry.server.zk.util;

import com.glodon.paas.framework.registry.core.status.ClientStatus;
import com.glodon.paas.framework.registry.core.status.ServiceStatus;
import com.glodon.paas.framework.registry.core.type.ClientBasic;
import com.glodon.paas.framework.registry.core.type.ServiceBasic;
import com.glodon.paas.framework.registry.server.zk.watcher.ClientWatcher;
import com.glodon.paas.framework.registry.server.zk.watcher.ServiceWatcher;
import org.apache.commons.lang.SerializationUtils;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.util.List;

/**
 * @author Don Li
 */
public class ZooKeeperServerUtils {

    private static final Logger LOGGER = LoggerFactory.getLogger(ZooKeeperServerUtils.class);

    public static void setObject(ZooKeeper zk, String path, Serializable object, int version) throws KeeperException, InterruptedException {
        zk.setData(path, SerializationUtils.serialize(object), version);
    }

    public static Object getObject(ZooKeeper zk, String path) throws KeeperException, InterruptedException {
        return SerializationUtils.deserialize(zk.getData(path, false, new Stat()));
    }

    public static synchronized void updateServiceStatus(ZooKeeper zk, String servicePath) throws KeeperException, InterruptedException {
        Stat serviceStat = new Stat();
        ServiceBasic serviceBasic = (ServiceBasic) SerializationUtils.deserialize(zk.getData(servicePath, false, serviceStat));
        // watch the changes of clients
        List<String> clients = zk.getChildren(servicePath, new ServiceWatcher(zk));
        if (serviceBasic.getStatus() == ServiceStatus.AVAILABLE) { // if an available service has no client or no client is up, then mark it as unavailable
            if (clients == null || clients.size() == 0) {
                markServiceAsUnavailable(zk, servicePath, serviceStat, serviceBasic);
                return;
            }
            int numOfAvailableClient = serviceStat.getNumChildren();
            for (String client : clients) {
                final String clientPath = servicePath + "/" + client;
                Stat clientStat = new Stat();
                ClientBasic clientBasic = (ClientBasic) SerializationUtils.deserialize(zk.getData(clientPath, new ClientWatcher(zk), clientStat));
                if (clientBasic.getClientStatus() != ClientStatus.UP) {
                    numOfAvailableClient--;
                }
            }
            if (numOfAvailableClient == 0) {
                markServiceAsUnavailable(zk, servicePath, serviceStat, serviceBasic);
            } else if (numOfAvailableClient != serviceBasic.getAvailableInstances()) {
                updateNumOfAvailableInstance(zk, servicePath, serviceStat, serviceBasic, numOfAvailableClient);
            }
        } else if (serviceBasic.getStatus() == ServiceStatus.UNAVAILABLE) { // if an unavailable service has at lease one up client, then mark it as available
            int numOfAvailableClient = 0;
            if (clients != null && clients.size() > 0) {
                for (String client : clients) {
                    final String clientPath = servicePath + "/" + client;
                    Stat clientStat = new Stat();
                    ClientBasic clientBasic = (ClientBasic) SerializationUtils.deserialize(zk.getData(clientPath, new ClientWatcher(zk), clientStat));
                    if (clientBasic.getClientStatus() == ClientStatus.UP) {
                        numOfAvailableClient++;
                    }
                }
            }
            if (numOfAvailableClient > 0) {
                markServiceAsAvailable(zk, servicePath, serviceStat, serviceBasic, numOfAvailableClient);
            }
        }
    }

    private static void markServiceAsAvailable(ZooKeeper zk, String servicePath, Stat serviceStat, ServiceBasic serviceBasic, int numOfAvailableClient) throws KeeperException, InterruptedException {
        serviceBasic.setStatus(ServiceStatus.AVAILABLE);
        serviceBasic.setAvailableInstances(numOfAvailableClient);
        zk.setData(servicePath, SerializationUtils.serialize(serviceBasic), serviceStat.getVersion());
    }

    private static void markServiceAsUnavailable(ZooKeeper zk, String servicePath, Stat serviceStat, ServiceBasic serviceBasic) throws KeeperException, InterruptedException {
        serviceBasic.setStatus(ServiceStatus.UNAVAILABLE);
        serviceBasic.setAvailableInstances(0);
        zk.setData(servicePath, SerializationUtils.serialize(serviceBasic), serviceStat.getVersion());
    }

    private static void updateNumOfAvailableInstance(ZooKeeper zk, String servicePath, Stat serviceStat, ServiceBasic serviceBasic, int numOfAvailableClient) throws KeeperException, InterruptedException {
        serviceBasic.setAvailableInstances(numOfAvailableClient);
        zk.setData(servicePath, SerializationUtils.serialize(serviceBasic), serviceStat.getVersion());
    }
}
